package fun.yao.common.nio;

import fun.yao.common.Constant;
import fun.yao.common.monitor.MonitorThread;
import fun.yao.common.security.SSLConstant;
import fun.yao.common.timmer.IGessTimerTask;
import fun.yao.common.uitls.LogUtil;

import java.io.IOException;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 批量SocketChannel管理线程
 *
 */
public final class SocketManagerThread extends Thread implements IGessTimerTask
{
    /**选择器*/
    private Selector mSelector = null;

    /**当前线程管理的SocketChannel集合*/
    private Map<SocketChannel,GessSocketChannel> mapGessSocketChannel = new ConcurrentHashMap<SocketChannel,GessSocketChannel>(200);

    /**读取一条完整报文后的回调方法*/
    private Method callReadFullMsgMethod = null;
    /**读取一条完整报文后的回调对象*/
    private Object callReadFullMsgObj = null;

    /**Socket关闭事件的回调方法*/
    private Method callSocketClosedMethod = null;
    /**Socket关闭事件的回调对象*/
    private Object callSocketClosedObj = null;

    /**是否运行当前线程*/
    private boolean bIsRun = true;

    /**读取的总字节数*/
    private long mReadBytes = 0;
    /**写入的总字节数*/
    private long mWriteBytes = 0;
    /**计量的起启时间*/
    private long mStartTime = 0;

    /** 待进行事件变更的SocketChannel add by csl 2008.11.24 第一个元素：SocketChannel 第二个元素：ops*/
    private ConcurrentLinkedQueue<Object[]> cqSckChlEventChange = new ConcurrentLinkedQueue<Object[]>();

    /** 待进行添加的Socket队列 */
    private ConcurrentLinkedQueue<SocketChannel> cqSckChlWaitAdd = new ConcurrentLinkedQueue<SocketChannel>();
    /** 是否需要添加Socket */
    private volatile boolean bIsNeedAddSck = false;

    /** 报文通讯头的长度 */
    private int mHeadLen = Constant.MSG_LEN_LEN;
    /**是否是短连接，即只接收一个单包的请求，只发送一个单包*/
    private boolean bIsShortLink = false;

    /**是否需要压缩/解压缩报文 add by csl 2009.5.21 */
    private boolean bIsNeedZip = false;

    /** 报文通讯头的长度是否使用字节表示 add by csl 2010.5.18 */
    private boolean bIsUseByteHeadLen = false;

    //---------------------------------------------------------- 加密加关 add by csl 2008.11.11
    /** 加密模式，默认不加密 */
    private byte mEncryptMode = SSLConstant.ENCRYPT_MODE_NO ;
    /** 会话ID */
    private String mSessionId = "";

    /**socket read 接口*/
    private IGessSocketRead gessSocketRead = null;

    public SocketManagerThread() throws IOException
    {
        bIsRun = true;
        mSelector = Selector.open();
    }

    public SocketManagerThread(int vHeadLen) throws IOException
    {
        bIsRun = true;
        mSelector = Selector.open();
        mHeadLen = vHeadLen;
    }

    /**
     * 设置socket read
     * @param gessSocketRead
     */
    public void setGessSocketRead(IGessSocketRead gessSocketRead)
    {
        this.gessSocketRead = gessSocketRead;
    }

    /**
     * 设置是否是短连接
     * @param flag true:短连接 false:长连接
     */
    public void setIsShortLink(boolean flag)
    {
        this.bIsShortLink = flag;
    }

    /**
     * 设置是否需要压缩/解压缩报文
     * @param flag
     */
    public void setIsNeedZip(boolean flag)
    {
        this.bIsNeedZip = flag;
    }

    /**
     * 设置加密信息
     * @param iEncryptMode  加密模式
     * @param sSessionId    会话ID
     */
    public void setEncryptInfo(byte iEncryptMode,String sSessionId)
    {
        this.mEncryptMode = iEncryptMode;
        this.mSessionId   = sSessionId;
    }

    /**
     * 设置报文的通讯长度是否使用字节表示
     * @param flag
     */
    public void setIsUseByteHeadLen(boolean flag)
    {
        this.bIsUseByteHeadLen = flag;
    }

    //---------------------------------------------------------- 回调函数
    /**
     * 设置读取一条完整报文后的回调方法。回调方法应尽可能的快速处理
     * 回调方法原形为：method(SocketChannel vSocketChannel, ByteBuffer vFullMsgBuff);
     * @param objCall     接收一条完整报文的回调方法的实例
     * @param sMethodName  接收一条完整报文的回调方法的名称
     */
    public void regReadFullMsgMethod(Object objCall, String sMethodName)
            throws NoSuchMethodException,SecurityException
    {
        this.callReadFullMsgMethod = objCall.getClass().getMethod(sMethodName,new Class[]{SocketChannel.class,ByteBuffer.class});
        this.callReadFullMsgObj = objCall;
    }

    /**
     * 设置Socket关闭后的回调方法
     * @param objCall      接收Socket关闭事件的回调方法实例
     * @param sMethodName  接收Socket关闭事件的回调方法名称
     * 回调方法原形为：method(SocketChannel vSocketChannel);
     * @throws NoSuchMethodException
     * @throws SecurityException
     */
    public void regSocketClosed(Object objCall, String sMethodName)
            throws NoSuchMethodException,SecurityException
    {
        this.callSocketClosedMethod = objCall.getClass().getMethod(sMethodName,new Class[]{SocketChannel.class});
        this.callSocketClosedObj = objCall;
    }

    /**
     * 添加待管理的SocketChannel对象
     * @param vSocketChannel
     * @return
     */
    public GessSocketChannel addSocketChannel(SocketChannel vSocketChannel) throws Exception
    {
        GessSocketChannel gsc = null;
        try
        {
            //配置非阻塞方式
            vSocketChannel.configureBlocking(false);

            //生成二级系统的SocketChannel
            gsc = new GessSocketChannel(this.mSelector,vSocketChannel,this.mHeadLen);
            gsc.setIsShortLink(this.bIsShortLink);
            gsc.setIsNeedZip(this.bIsNeedZip);
            gsc.setEncryptInfo(this.mEncryptMode,this.mSessionId);
            gsc.setIsUseByteHeadLen(this.bIsUseByteHeadLen);
            gsc.regSocketClosedMethod(this,"closeSocketChannel");
            gsc.regReadFullMsgMethod(this,"readFullMsg");
            gsc.setWriteByEvent(true,this.cqSckChlEventChange);//add by csl 2008.11.24
            gsc.setGessSocketRead(gessSocketRead);

            //添加到集合中
            mapGessSocketChannel.put(vSocketChannel,gsc);

            //添加到队列中，注册读事件
            //gsc.registerEvent(SelectionKey.OP_READ);
            this.cqSckChlWaitAdd.add(vSocketChannel);
            this.bIsNeedAddSck = true;
            this.mSelector.wakeup();

            //CommUtil.WriteLog(Constant.DEBUG,"添加" + vSocketChannel.socket().toString());


        }catch(Exception cce){
            this.closeSocketChannel(vSocketChannel);
            LogUtil.WriteLog(Constant.NORMAL_ERROR,cce);
        }

        return gsc;

    }

    /**
     * 添加Socket
     */
    private void loadSckChlAdd()
    {
        for(;;)
        {
            SocketChannel sck = this.cqSckChlWaitAdd.poll();
            if ( sck != null )
            {
                try
                {

                    sck.register(this.mSelector, SelectionKey.OP_READ);
                    //CommUtil.writeLog("Socket", Constant.INFO, sck.socket().toString() + ":成功注册事件[" + SelectionKey.OP_READ + "]");
                }catch(Exception cce){
                    if ( sck != null )
                        this.closeSocketChannel(sck);
                    LogUtil.WriteLog(Constant.NORMAL_ERROR,cce);
                }
            }else{
                break;
            }
        }
        bIsNeedAddSck = false;
    }

    /**
     * 装载SocketChannel 事件改变
     *
     */
    private void loadSckChlEvtChange()
    {
        for(;;)
        {
            Object[] objEvt = this.cqSckChlEventChange.poll();
            if ( objEvt != null )
            {
                SocketChannel sc = null;
                try{
                    sc = (SocketChannel)objEvt[0];
                    Integer intOps = (Integer)objEvt[1];
                    if ( sc != null && sc.isOpen() == true )
                    {
                        sc.register(this.mSelector, intOps.intValue());
                        //CommUtil.writeLog("Socket", Constant.INFO, sc.socket().toString() + ":成功注册事件[" + intOps.intValue() + "]");
                    }
                }catch(Exception cce){
                    if ( sc != null )
                        this.closeSocketChannel(sc);
                    LogUtil.WriteLog(Constant.NORMAL_ERROR,cce);
                }
            }else
                break;
        }
    }

    /**
     * 移除管理的SocketChannel对象
     * @param vSocketChannel
     */
	/*
	public void removeSocketChannel(SocketChannel vSocketChannel) throws Exception
	{
		//vSocketChannel.register(this.mSelector,0);
		this.cqSckChlEventChange.add(new Object[]{vSocketChannel,0});
		this.mSelector.wakeup();
		mapGessSocketChannel.remove(vSocketChannel);
		CommUtil.WriteLog(Constant.DEBUG,"移除" + vSocketChannel.socket().toString());
	}*/

    /**
     * 线程方法
     */
    public void run()
    {
        String sName = "SocketManagerThread-" + this.getName();
        mStartTime = System.currentTimeMillis();
        while ( bIsRun)
        {
            try{
                int num = mSelector.select(2000);
                try
                {
                    MonitorThread.addCycleInvoke(sName);
                    if ( num > 0 )
                    {
                        Iterator<SelectionKey> it = mSelector.selectedKeys().iterator();

                        while (it.hasNext())
                        {
                            //快速处理Socket的注册 add by csl 2008.12.2
                            if ( bIsNeedAddSck )
                                this.loadSckChlAdd();

                            //处理事件
                            SelectionKey s = (SelectionKey)it.next();
                            it.remove();
                            doEvent(s);
                        }
                    }

                    //装载SocketChannel事件改变消息 2008.11.24
                    if ( this.cqSckChlEventChange.peek() != null )
                        this.loadSckChlEvtChange();

                    //添加Socket
                    if ( this.cqSckChlWaitAdd.peek() != null )
                        this.loadSckChlAdd();

                }catch(Exception ee)
                {
                    LogUtil.WriteLog(Constant.BADLY_ERROR,ee);
                }
            }catch(Exception e)
            {
                LogUtil.WriteLog(Constant.BADLY_ERROR,e);
            }
        }
    }

    /**处理事件*/
    private void doEvent(SelectionKey s)
    {
        try
        {

            // 读取事件
            if (s.isValid() && s.isReadable() )
            {
                SocketChannel sc = (SocketChannel) s.channel();
                GessSocketChannel gsc = (GessSocketChannel)this.mapGessSocketChannel.get(sc);
                if ( gsc != null )
                {
                    int num = gsc.doRead();
                    statisticsReadDataStream(num);
                }
            }

            // 写入事件
            if (s.isValid() && s.isWritable() )
            {
                SocketChannel sc = (SocketChannel) s.channel();
                GessSocketChannel gsc = this.mapGessSocketChannel.get(sc);
                if ( gsc != null )
                {
                    int num = gsc.doWrite();
                    statisticsWriteDataStream(num);
                }
            }

        }catch(Exception e)
        {
            LogUtil.WriteLog(Constant.BADLY_ERROR,e);
        }
    }

    /**添加待发送报文*/
    public void addWriteMsg(SocketChannel vSocketChannel , StringBuffer sbMsg)
    {
        GessSocketChannel gsc = this.mapGessSocketChannel.get(vSocketChannel);
        if ( gsc != null )
        {
            gsc.addWriteMsg(sbMsg);
        }else
            LogUtil.WriteLog(Constant.BADLY_ERROR,"未找到可发送的SocketChannel[" + vSocketChannel.socket().toString() +  "]");
    }

    //重载
    public void addWriteMsg(SocketChannel vSocketChannel, ByteBuffer msgBuff) throws Exception
    {
        GessSocketChannel gsc = this.mapGessSocketChannel.get(vSocketChannel);
        if ( gsc != null )
            gsc.addWriteMsg(msgBuff);
        else
            LogUtil.WriteLog(Constant.BADLY_ERROR,"未找到可发送的SocketChannel");
    }

    /**添加待发送报文*/
    public void addWriteMsg(StringBuffer sbMsg)
    {
        SocketChannel scIdlesse = this.getIdlesseSocketChannel();
        if ( scIdlesse != null )
            this.addWriteMsg(scIdlesse,sbMsg);
        else
            LogUtil.WriteLog(Constant.BADLY_ERROR,"未找到可发送的SocketChannel");
    }

    //重载
    public void addWriteMsg(ByteBuffer msgBuff) throws Exception
    {
        SocketChannel scIdlesse = this.getIdlesseSocketChannel();
        if ( scIdlesse != null )
            this.addWriteMsg(scIdlesse,msgBuff);
        else
            LogUtil.WriteLog(Constant.BADLY_ERROR,"未找到可发送的SocketChannel");
    }


    /**
     * 读取完整报文
     * @param vSocketChannel
     */
    public void readFullMsg(SocketChannel vSocketChannel , ByteBuffer msgBuff)
    {
        try{
            if ( this.callReadFullMsgObj != null && this.callReadFullMsgMethod != null )
            {
                this.callReadFullMsgMethod.invoke(this.callReadFullMsgObj,new Object[]{vSocketChannel,msgBuff});
            }
        }catch(Exception e){
            LogUtil.WriteLog(Constant.NORMAL_ERROR,e);
        }
    }

    /**
     * 关闭Socket管理服务
     */
    public void close()
    {
        try
        {

            bIsRun = false;

            Iterator<SocketChannel> it = this.mapGessSocketChannel.keySet().iterator();
            while (it.hasNext())
            {
                SocketChannel sc = (SocketChannel)it.next();
                sc.close();
            }

            if ( this.mSelector != null )
                this.mSelector.close();

            LogUtil.writeSystemLog(Constant.NOTICE,"关闭通道管理线程完成");
        }catch(Exception e)
        {
            LogUtil.WriteLog(Constant.NORMAL_ERROR,e);
        }
    }

    /**
     * 关闭SocketChannel
     */
    public void closeSocketChannel(SocketChannel vSocketChannel)
    {
        try
        {
            this.mapGessSocketChannel.remove(vSocketChannel);

            if ( this.callSocketClosedObj != null && this.callSocketClosedMethod != null )
            {
                this.callSocketClosedMethod.invoke(this.callSocketClosedObj,new Object[]{vSocketChannel});
            }
            //CommUtil.WriteLog(Constant.DEBUG,"关闭" + vSocketChannel.socket().toString());
        }catch(Exception e){
            LogUtil.WriteLog(Constant.NORMAL_ERROR,e);
        }
    }

    /**获得当前比较空闲的SocketChannel*/
    public SocketChannel getIdlesseSocketChannel()
    {
        GessSocketChannel gscIdlesse = null;

        Iterator<SocketChannel> it = this.mapGessSocketChannel.keySet().iterator();

        while (it.hasNext())
        {
            SocketChannel tmpSc = (SocketChannel)it.next();

            GessSocketChannel tmpGsc = this.mapGessSocketChannel.get(tmpSc);

            if ( gscIdlesse == null || gscIdlesse.getUseRate() > tmpGsc.getUseRate() )
            {
                gscIdlesse = tmpGsc;
            }
        }

        if ( gscIdlesse == null )
            return null;
        else
            return gscIdlesse.getSocketChannel() ;
    }

    /**
     * 获得正在写的Socket数量
     * @return
     */
    public synchronized int getWriteSocketChannelNum()
    {
        int num = 0 ;
        Iterator<SocketChannel> iter =  this.mapGessSocketChannel.keySet().iterator();
        while ( iter.hasNext() )
        {
            GessSocketChannel gsc = this.mapGessSocketChannel.get(iter.next());

            //判别是否正在发送
            if(gsc.isSending())
            {
                num++;
            }

        }
        return num;
    }

    /**
     * 获得管理的SocketChannel个数
     * @return
     */
    public int getSocketChannelCount()
    {
        return this.mapGessSocketChannel.size();
    }

    /**获得管理的Socket集合*/
    public Map<SocketChannel,GessSocketChannel> getManagerSocket()
    {
        return this.mapGessSocketChannel;
    }

    /**
     * 根据API的SocketChannel返回二级系统对应的GessSocketChannel
     * @param sc
     * @return
     */
    public GessSocketChannel getGessSocketChannel(SocketChannel sc)
    {
        return this.mapGessSocketChannel.get(sc);
    }

    /**
     * 获得当前SocketChannel读取的总字节数
     * @return
     */
    public long getSumReadBytes()
    {
        return this.mReadBytes;
    }

    /**
     * 获得当前SocketChannel写入的总字节数
     * @return
     */
    public long getSumWriteBytes()
    {
        return this.mWriteBytes;
    }

    /**
     * 获得当前Socket的利用率 （读入总字节＋写入总字节）/套接字连接时间（秒）
     * @return
     */
    public double getUseRate()
    {
        long currTime = System.currentTimeMillis();
        long invTime = currTime - this.mStartTime;
        if ( invTime > Constant.CFG_COLLECT_DATA_STREAM_SIZE_TIME )
        {
            this.mStartTime = currTime;
            this.mWriteBytes = 0;
            this.mReadBytes  = 0;
        }

        invTime = invTime/1000;

        long bytes = this.mReadBytes + this.mWriteBytes;
        if ( invTime <= 0 )
            return bytes;
        else
            return bytes*1.0/(invTime*1.0);
    }

    /**
     * 统计输入流量
     * @param size     当前流大小
     */
    private void statisticsReadDataStream(int size)
    {
        long currTime = System.currentTimeMillis();
        long invTime = currTime - this.mStartTime;
        if ( invTime > Constant.CFG_COLLECT_DATA_STREAM_SIZE_TIME )
        {
            this.mStartTime = currTime;
            this.mWriteBytes = 0;
            this.mReadBytes  = 0;
        }

        //CommUtil.WriteLog(1,"w=" + this.mWriteBytes + ",r=" + this.mReadBytes + ",rate=" + this.getUseRate() );

        this.mReadBytes += size;
    }

    /**
     * 统计输出流量
     * @param size     当前流大小
     */
    private void statisticsWriteDataStream(int size)
    {
        long currTime = System.currentTimeMillis();
        long invTime = currTime - this.mStartTime;
        if ( invTime > Constant.CFG_COLLECT_DATA_STREAM_SIZE_TIME )
        {
            this.mStartTime = currTime;
            this.mWriteBytes = 0;
            this.mReadBytes  = 0;
        }

        //CommUtil.WriteLog(1,"w=" + this.mWriteBytes + ",r=" + this.mReadBytes + ",rate=" + this.getUseRate() );

        this.mWriteBytes+= size;
    }

    public void onTimeOut()
    {
        Map<SocketChannel,GessSocketChannel> tempMapGessSocketChannel = new ConcurrentHashMap<SocketChannel, GessSocketChannel>();
        tempMapGessSocketChannel.putAll(this.mapGessSocketChannel);
        Iterator<SocketChannel> it = tempMapGessSocketChannel.keySet().iterator();
        while(it.hasNext())
        {
            SocketChannel sockeChannel = it.next();
            GessSocketChannel gessSocketChannel = tempMapGessSocketChannel.get(sockeChannel);
            //使用checkCanSendTestMsg 判断是否超时(客户端未发送心跳包，服务端未推送任何报文)
            if (gessSocketChannel != null &&  gessSocketChannel.checkCanSendTestMsg(2*4) )
            {
                LogUtil.writeSystemLog(Constant.NOTICE,"接收客户端[" + sockeChannel + "]消息超时，关闭连接！");
                this.closeSocketChannel(sockeChannel);
            }
        }
    }
}
